package cn.itcast.flink.connector;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author lilulu
 */
//案例演示：使用JDBC Sink 连接器，将数据保存至MySQL表中，继承RichSinkFunction
public class ConnectorJdbcSinkDemo {

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    private static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Student> studentDataStreamSource = env.fromElements(
                new Student(34, "zhaoliu", 19),
                new Student(35, "wangwu", 20),
                new Student(36, "zhaoliu", 19)
        );
        // 3. 数据转换-transformation
        // 4. 数据终端-sink
        // a. 插入数据
        // b. 设置语句中占位的值
        // c. 插入数据时属性设置
        // d. 数据库连接属性设置
        SinkFunction<Student> jdbcSink = JdbcSink.sink(
                "INSERT INTO db_flink.t_student(id, name, age) VALUES (?, ?, ?)",
                new JdbcStatementBuilder<Student>() {
                    public void accept(PreparedStatement preparedStatement, Student student) throws SQLException {
                        preparedStatement.setInt(1, student.id);
                        preparedStatement.setString(2, student.name);
                        preparedStatement.setInt(3, student.age);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchIntervalMs(200)
                        .withBatchSize(1000)
                        .withMaxRetries(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUrl("jdbc:mysql://node1.itcast.cn:3306/?useSSL=false")
                        .withUsername("root")
                        .withPassword("123456")
                        .withConnectionCheckTimeoutSeconds(5000)
                        .build()
        );
        // 4-1. 创建Sink接收器对象
        studentDataStreamSource.addSink(jdbcSink);
        // 4-2. 添加接收器
        // 5. 触发执行-execute
        env.execute("ConnectorJdbcSinkDemo");
    }
}